package org.javaboy.msg.queue;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.javaboy.msg.model.Msg;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.List;
import java.util.UUID;

/**
 * 延迟消息队列
 */
public class DelayMsgQueue {
    private Jedis jedis;
    private String queueKey;

    public DelayMsgQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    /**
     * 消息入队
     *
     * @param data
     */
    public void send(Object data, long delaySeconds) {
        // 发送消息
        Msg msg = new Msg();
        msg.setId(UUID.randomUUID().toString());
        msg.setData(data);
        try {
            String s = new ObjectMapper().writeValueAsString(msg);
            //发送延迟消息
            jedis.zadd(queueKey, System.currentTimeMillis() + delaySeconds * 1000L, s);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    public void receive() {
        // 接收消息
        // Thread.interrupted() 返回当前线程的中断状态
        while (!Thread.interrupted()) {
            //读消息，读 score 从 0 到当前时间的消息
            List<String> list = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis());
            if (list.isEmpty()) {
                //说明没有消息
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }
            // 获取消息
            for (String msg : list) {
                // 消费之前，先删除消息，防止重复消费
                if (jedis.zrem(queueKey, msg) > 0) {
                    try {
                        Msg value = new ObjectMapper().readValue(msg, Msg.class);
                        System.out.println(new Date() + " >>> " + Thread.currentThread().getName() + "消费了消息：" + value);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                }
            }

        }
    }

}
